package com.jie.netty.processor;

import com.jie.constants.RedisConstants;
import com.jie.entity.Notice;
import com.jie.netty.im.UserChannelCtxMap;
import com.jie.netty.message.NoticeInfoMessage;
import com.jie.netty.model.IMRecvInfo;
import com.jie.netty.model.IMSendInfo;
import com.jie.netty.type.IMCmdType;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.lang.reflect.Field;
import java.util.*;

import static com.jie.constants.RedisConstants.USER_LIST;

/**
 * @Classname NotifyMessageProcessor 公共通知业务逻辑
 * @Author: 杰
 * @Project :net-disk
 * @DateTime: 2024/7/2 14:31
 **/
@Slf4j
@Component
public class NotifyMessageProcessor extends AbstractMessageProcessor<NoticeInfoMessage>{
    @Autowired
    private RedisTemplate redisTemplate;
    /**
     * 消息发送逻辑
     * @param
     */
    @Override
    public void process(NoticeInfoMessage message) {
        //给当前全部用户推送消息 不在线的上线拉取
        Set<String> userList = redisTemplate.opsForSet().members(USER_LIST);
        //找到他们的channel
        Notice notice = message.getData();
        //这里抽出来一个List做用户已读列表然后管道存入redis的count+1
        List<String> userReadList = new ArrayList<>();
        try {
            log.info("信息{}",notice);
            for (String username : userList) {
                ChannelHandlerContext channelCtx = UserChannelCtxMap.getChannelCtx(username, 0);
                if (channelCtx != null) {
                    // 推送消息到用户
                    IMSendInfo sendInfo = new IMSendInfo();
                    sendInfo.setCmd(IMCmdType.NOTIFY_MESSAGE.code());
                    sendInfo.setData(message);
                    channelCtx.channel().writeAndFlush(sendInfo);
                    //并且把已读用户加入到redis中，让用户已读的消息+1
                    userReadList.add(username);
                } else {
                    log.info("该{}用户不在线,后续上线再做拉取处理", username);
                }
            }
            //这里将用户的已读消息的count+1
            if (userReadList.size() > 0) {
                addNotifyCountWithLua(userReadList);
            }
        } catch (Exception e) {
            // 消息推送失败确认
            log.error("公共发送异常,内容:{}",message, e);
        }
    }

    /**
     * 通知用户已读消息数量+1
     * @param usernames
     */
    @Async
    public void addNotifyCountWithLua(List<String> usernames) {
        DefaultRedisScript<Long> script = new DefaultRedisScript<>("local count = tonumber(redis.call('HGET', KEYS[1], ARGV[1])) "
                +
                "if count == nil then " +
                "   redis.call('HSET', KEYS[1], ARGV[1], 1) " +
                "else " +
                "   redis.call('HSET', KEYS[1], ARGV[1], count + 1) " +
                "end", Long.class);

        List<String> keys = new ArrayList<>();
        keys.add(RedisConstants.NOTICE_LIST);
        // 创建 Redis 管道
        RedisCallback<List<Object>> pipelineCallback = connection -> {
            for (String user : usernames) {
                redisTemplate.execute(script, keys, user);
            }
            return null; // 回调函数必须返回 null
        };
        // 执行管道操作
        redisTemplate.executePipelined(pipelineCallback);
    }
}
